//package com.example.kafka.flink;
//
//import com.alibaba.fastjson.JSONArray;
//
//import com.fasterxml.jackson.databind.JsonNode;
//import com.fasterxml.jackson.databind.ObjectMapper;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.lang3.StringUtils;
//import org.apache.flink.api.common.functions.FlatMapFunction;
//import org.apache.flink.api.common.serialization.SimpleStringSchema;
//import org.apache.flink.api.java.functions.KeySelector;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
//import org.apache.flink.table.api.Table;
//import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
//import org.springframework.boot.CommandLineRunner;
//import org.springframework.kafka.support.JacksonUtils;
//import org.springframework.stereotype.Component;
//
//import java.util.ArrayList;
//import java.util.List;
//import java.util.Properties;
//
//@Slf4j
//@Component
//public class AnalyseMetricByFlink implements CommandLineRunner {
//
//    @Override
//    public void run(String... args) throws Exception {
//        // 创建执行环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        // 设置并行度1
//        env.setParallelism(1);
//
//        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//
//        Properties properties = new Properties();
//        properties.setProperty("bootstrap.servers", "60.60.60.209:9092");
//        // 下面这些次要参数
//        properties.setProperty("group.id", "nrMetric01");
//        properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//        properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//        properties.setProperty("auto.offset.reset", "latest");
//
//        // flink添加外部数据源
//        DataStream<String> dataStream = env.addSource(new FlinkKafkaConsumer<>("nr_metric_result", new SimpleStringSchema(), properties));
//        DataStream<JSONArray> mapStream = dataStream.keyBy(new DemoKeySelector()).map(JSONArray::parseArray);
//        SingleOutputStreamOperator<String> flatMapStream = mapStream.flatMap((FlatMapFunction<JSONArray, String>) (jsonArray, out) -> {
//            List<String> list = jsonArray.toJavaList(String.class);
//            for (String field : list) {
//                out.collect(field);
//            }
//        });
//        Table table = tableEnv.fromDataStream(flatMapStream);
//        tableEnv.createTemporaryView("InputTable", table);
//
//        Table dirResult = tableEnv.sqlQuery(
//                "SELECT * FROM InputTable WHERE metric = 'proc.start.dir'");
//
//        Table typeResult = tableEnv.sqlQuery(
//                "SELECT * FROM InputTable WHERE metric = 'proc.type'");
//
//        List<String> list  = new ArrayList<>();
////        ProcessStandingBookModel processStandingBookModel;
//        // 打印输出
////        tableEnv.toAppendStream(result, Row.class).print("dev");
//        tableEnv.toAppendStream(dirResult, String.class).map(a -> {
////            processStandingBookModel = new ProcessStandingBookModel();
////            processStandingBookModel.setName(matchComponent(a.getTags(),"component"));
////            processStandingBookModel.setNameAlias(matchComponent(a.getTags(),"componentalias"));
////            processStandingBookModel.setDir(a.getValue());
////            list.add(processStandingBookModel);
//            return null;
//        });
//        tableEnv.toAppendStream(typeResult, String.class).map(a -> {
//            list.forEach(l->{
//
//            });
//            return null;
//        });
//        env.execute();
//    }
//
//    public static class DemoKeySelector implements KeySelector<String, String> {
//        private transient ObjectMapper objectMapper;
//        public static final Object objectMapperInitLock = new Object();
//
//        @Override
//        public String getKey(String value) throws Exception {
//            if (objectMapper == null) {
//                synchronized (objectMapperInitLock) {
//                    this.objectMapper = JacksonUtils.enhancedObjectMapper();
//                }
//            }
//
//            JsonNode jsonNode = objectMapper.readTree(value);
//            String key = findStringValue(jsonNode, "");
//            if (StringUtils.isBlank(key)) {
//                return "";
//            } else {
//                return key;
//            }
//        }
//    }
//
//    private static String findStringValue(JsonNode jsonNode, String fieldName) {
//        if (jsonNode == null) {
//            return null;
//        }
//
//        JsonNode node = jsonNode.get(fieldName);
//        return node == null ? null : node.asText();
//    }
//}
